package com.tkafka.confgi;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * @author: LCG
 * 作者: LCG
 * 日期: 2021/5/24  16:07
 * 描述:
 */
@Configuration
public class MyKafkaConfig {

    @Bean("myKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory getKafkaListenerContainerFactory(ConsumerFactory<String,String> consumerFactory){

        ConcurrentKafkaListenerContainerFactory concurrentKafkaListenerContainerFactory=
                new ConcurrentKafkaListenerContainerFactory();

        concurrentKafkaListenerContainerFactory.setConsumerFactory(consumerFactory);
         //配置 topic 主题（队列）
        concurrentKafkaListenerContainerFactory.createContainer("HHHH");
         //配置消息确认自动
        concurrentKafkaListenerContainerFactory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        concurrentKafkaListenerContainerFactory.getContainerProperties().setGroupId("TT");


        return concurrentKafkaListenerContainerFactory;

    }

}
